package com.example.demo;

import jakarta.annotation.PostConstruct;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Component
public class RocketMQConsumer {

    private static final Logger log = LoggerFactory.getLogger(RocketMQConsumer.class);

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        // 创建消费者，并指定消费组
        consumer = new DefaultMQPushConsumer("test-consumer-group");

        // 设置 NameServer 地址
        consumer.setNamesrvAddr("192.168.10.32:9876");

        // 订阅主题
        consumer.subscribe("TestTopic", "*");
        consumer.setMaxReconsumeTimes(1);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 设置并发消费监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String messageBody = new String(msg.getBody(), StandardCharsets.UTF_8);
                    log.info("收到消息: {}", messageBody);

                    // 模拟伪消费，直接返回 RECONSUME_LATER 让消息重试
                    if (messageBody.contains("retry")) {
                        log.warn("伪消费消息: {}, 进行重试", messageBody);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    // 正常消费
                    log.info("成功消费消息: {}", messageBody);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        log.info("RocketMQ 消费者已启动");
    }
}

